package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.api.R;
import com.heima.common.constant.GlobalConstant;
import com.heima.common.constant.ScheduleConstants;
import com.heima.model.common.dtos.ResponseResult;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.service.TaskService;
import com.heima.schedule.service.TaskinfoLogsService;
import com.heima.schedule.service.TaskinfoService;
import com.xxl.job.core.handler.annotation.XxlJob;
import jodd.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.statement.select.KSQLWindow;
import org.checkerframework.checker.units.qual.A;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.*;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
@Slf4j
@Service
@Transactional
public class TaskServiceImpl implements TaskService {
    @Autowired
    TaskinfoService taskinfoService;
    @Autowired
    TaskinfoLogsService taskinfoLogsService;
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    RedissonClient redissonClient;
    @Override
    public ResponseResult addTask(Task task) {
        if(!addTaskToDB(task))
        {
            return ResponseResult.errorResult(AppHttpCodeEnum.TASK_ADD_TO_DB_FAIL);
        }
        if(!addTaskToRedis(task))
        {
            return ResponseResult.errorResult(AppHttpCodeEnum.TASK_ADD_TO_DB_FAIL);
        }
        return ResponseResult.okResult(task.getTaskId());
    }

    private boolean addTaskToRedis(Task task) {
        try{
            String key=task.getTaskType()+"_"+task.getPriority();
            if(System.currentTimeMillis()>=task.getExecuteTime())
            {
                stringRedisTemplate.opsForList().leftPush(ScheduleConstants.TOPIC+key, JSON.toJSONString(task));
                return true;
            }
            Calendar calendar=Calendar.getInstance();
            calendar.add(Calendar.MINUTE,GlobalConstant.Z_SET_STORAGE_TIME);
            long timestamp=calendar.getTimeInMillis();
            if(timestamp>=task.getExecuteTime())
            {
                stringRedisTemplate.opsForZSet().add(ScheduleConstants.FUTURE+key, JSON.toJSONString(task),task.getExecuteTime());
                return true;
            }
            return true;
        }catch (Exception e) {
            return false;
        }
    }

    private boolean addTaskToDB(Task task) {
        try{
            Taskinfo taskinfo=new Taskinfo();
            TaskinfoLogs taskinfoLogs=new TaskinfoLogs();
            BeanUtils.copyProperties(task,taskinfo);
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskinfoService.save(taskinfo);

            task.setTaskId(taskinfo.getTaskId());
            BeanUtils.copyProperties(task,taskinfoLogs);
            taskinfoLogs.setVersion(GlobalConstant.VERSION_INIT);
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            taskinfoLogs.setExecuteTime(new Date(task.getExecuteTime()));
            taskinfoLogsService.save(taskinfoLogs);
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public ResponseResult cancelTask(long taskId) {
        try{
            //删除任务
            taskinfoService.removeById(taskId);
            //更新任务日志
            TaskinfoLogs taskinfoLogs = taskinfoLogsService.getById(taskId);
            taskinfoLogs.setStatus(ScheduleConstants.CANCELLED);
            taskinfoLogsService.updateById(taskinfoLogs);
            //得到task
            Task task=new Task();
            BeanUtils.copyProperties(taskinfoLogs,task);
            task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
            //删除redis中的task
            String key=task.getTaskType()+"_"+task.getPriority();
            String jsonTask=JSON.toJSONString(task);
            stringRedisTemplate.opsForZSet().remove(ScheduleConstants.FUTURE+key,jsonTask);
            stringRedisTemplate.opsForList().remove(ScheduleConstants.TOPIC+key,0,jsonTask);
            return ResponseResult.okResult();
        }catch (Exception e){
            return ResponseResult.errorResult(AppHttpCodeEnum.TASK_CANCEL_FAIL);
        }
    }

    @Override
    public ResponseResult pull(int type, int priority) {
        try{
            String key=type+"_"+priority;
            String json = stringRedisTemplate.opsForList().rightPop(ScheduleConstants.TOPIC + key);
            if(StringUtil.isNotBlank(json))
            {
                Task task = JSON.parseObject(json, Task.class);
                taskinfoService.removeById(task);
                TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
                taskinfoLogs.setTaskId(task.getTaskId());
                taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);
                taskinfoLogsService.updateById(taskinfoLogs);
                log.info("拉取任务成功:"+taskinfoLogs);
                return ResponseResult.okResult(task);
            }
            else return ResponseResult.okResult(null);
        }catch (Exception e){
            return ResponseResult.errorResult(AppHttpCodeEnum.FEIGN_FAIL);
        }
    }
    @PostConstruct
    public void init()
    {
        Thread redisZSetToListThread=new Thread(new Runnable() {
            @Override
            public void run() {
                while (true)
                {
                    redisZSetToList();
                }
            }
        });
        redisZSetToListThread.start();
    }
//    @Scheduled(cron = "0 */5 * * * ?")
    @XxlJob(GlobalConstant.XXL_JOB_BD_TO_REDIS_SCHEDULE)
    public void dbToRedisZSet()
    {
        log.info("db -> zset 执行开始...");
        RLock lock = redissonClient.getLock(GlobalConstant.LOCK_TAG + GlobalConstant.DB_TO_REDIS);
        if(lock.tryLock())
        {
            LocalDateTime now = LocalDateTime.now();
            LocalDateTime dateTime = now.plusMinutes(GlobalConstant.Z_SET_STORAGE_TIME);
            List<Taskinfo> taskinfoList = taskinfoService.list(Wrappers.<Taskinfo>lambdaQuery().le(Taskinfo::getExecuteTime, dateTime));
            for (Taskinfo taskinfo : taskinfoList) {
                String key=taskinfo.getTaskType()+"_"+taskinfo.getPriority();
                System.out.println("curr: "+System.currentTimeMillis());
                System.out.println("time: "+taskinfo.getExecuteTime().getTime());
                stringRedisTemplate.opsForZSet().add(ScheduleConstants.FUTURE+key,JSON.toJSONString(taskinfo),taskinfo.getExecuteTime().getTime());
            }
        }
        lock.unlock();
        log.info("db -> zset 执行结束");
    }
    public void redisZSetToList()
    {
//        log.info("zset -> list 执行开始...");
        List<String> keys= stringRedisTemplate.execute(new RedisCallback<List<String>>() {
            @Override
            public List<String> doInRedis(RedisConnection redisConnection) throws DataAccessException {
                List<String> list=new ArrayList<>();
                Cursor<byte[]> scan = redisConnection.scan(new ScanOptions.ScanOptionsBuilder()
                        .match(ScheduleConstants.FUTURE + "*")
                        .count(GlobalConstant.REDIS_SCAN_COUNT)
                        .build());
                while(scan.hasNext())
                {
                    list.add(new String(scan.next()) );
                }
                return list;
            }
        });
        List<String> list=new ArrayList<>();
        for (String key : keys) {
            Set<String> range = stringRedisTemplate.opsForZSet().rangeByScore(key, 0, System.currentTimeMillis());
            list.addAll(range);
        }
        if(list.isEmpty())
        {
            try {
                Thread.sleep(GlobalConstant.THREAD_SLEEP_TIME);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        else {
            log.info("redis zset->list :"+list);
            stringRedisTemplate.executePipelined(new RedisCallback<Object>() {
                @Override
                public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                    for (String s : list) {
                        Task task = JSON.parseObject(s, Task.class);
                        String key=task.getTaskType()+"_"+task.getPriority();
                        redisConnection.lPush((ScheduleConstants.TOPIC+key).getBytes(),s.getBytes());
                        redisConnection.zRem((ScheduleConstants.FUTURE+key).getBytes(),s.getBytes());

                    }
                    return null;
                }
            });
        }
//        log.info("zset -> list 执行结束");

    }

}
